package cmd

import (
	"context"
	"github.com/spf13/cobra"
	"log"
	"strings"
	kafka "github.com/segmentio/kafka-go"

)
var brokerList string
var topic string

var kafkaCmd = &cobra.Command{
	Use:   "kafka",
	Short: "kafka cluster check tool",
	Long: `kafka cluster check tool,it's will producers and consumers kafka topic'`,
	Run: func(cmd *cobra.Command, args []string)  {
		// Do Stuff Here
		log.Println("kafka command run")
		runKafkaCommand(brokerList,topic)
	},
}

func init() {
	kafkaCmd.Flags().StringVarP(&brokerList, "brokerlist", "b", "", "kafka cluster urls 逗号分隔")
	kafkaCmd.Flags().StringVarP(&topic, "topic", "t", "", "kafka topic")
	rootCmd.AddCommand(kafkaCmd)
}

func runKafkaCommand(brokerList string, topic string)  {
	writer := getKafkaWriter(brokerList, topic)
	defer writer.Close()

	err := writer.WriteMessages(context.Background(),kafka.Message{
		Value: []byte("ping"),
	})
	if err != nil {
		log.Fatal(err)
	}
	log.Println("ok")
}

func getKafkaWriter(kafkaURL, topic string) *kafka.Writer {
	brokers := strings.Split(kafkaURL, ",")
	log.Println("kafka writer brokers : ",kafkaURL)
	log.Println("kafka writer topic : ",topic)
	return kafka.NewWriter(kafka.WriterConfig{
		Brokers:  brokers,
		Topic:    topic,
		Balancer: &kafka.LeastBytes{},
	})
}


